package com.transport.analysis.traffic;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class TrafficDriver extends Configured implements Tool {

    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new TrafficDriver(), args);
        System.exit(res);
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = this.getConf();

        // Set database credentials for your Windows MySQL
        // You may need to replace these with your actual MySQL credentials
        conf.set("mapreduce.jdbc.driver.class", "com.mysql.cj.jdbc.Driver");
        conf.set("mapreduce.jdbc.url", "jdbc:mysql://192.168.177.1:3307/transport_data?serverTimezone=UTC");
        conf.set("mapreduce.jdbc.username", "root"); // Replace with your MySQL username
        conf.set("mapreduce.jdbc.password", "1234"); // Replace with your MySQL password

        // Specify the column to use for splitting the input
        conf.set("mapreduce.input.dbinputformat.input.orderby", "route_id");

        Job job = Job.getInstance(conf, "Intelligent Dispatching Suggestion Model Prototype");
        job.setJarByClass(TrafficDriver.class);

        // Configure DBInputFormat to read from the unified traffic_records_view
        // The Mapper will filter for bus records based on bus_id and subway_id
        String[] fields = { "bus_id", "route_id", "stop_id", "timestamp", "direction", "status", "passengers_on",
                "passengers_off", "subway_id" };

        // The SQL query to select data from the view.
        String dbQuery = "SELECT bus_id, route_id, stop_id, timestamp, direction, status, passengers_on, passengers_off, subway_id FROM traffic_records_view";

        DBInputFormat.setInput(job,
                TrafficRecordWritable.class,
                dbQuery,
                "SELECT COUNT(*) FROM traffic_records_view"); // Use a simple count query for the view

        job.setMapperClass(TrafficMapper.class);
        job.setReducerClass(TrafficReducer.class);

        // Set Mapper output key and value classes for Intelligent Dispatching
        // Suggestion Model Prototype
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        // Set Reducer output key and value classes for Intelligent Dispatching
        // Suggestion Model Prototype
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        // Set output path for this specific analysis
        FileOutputFormat.setOutputPath(job, new Path(args[0]));

        return job.waitForCompletion(true) ? 0 : 1;
    }
}